package com.azh.server.datam;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import com.azh.server.common.SubArray;
import com.azh.server.datam.dataItem.DataItem;
import com.azh.server.datam.logger.Logger;
import com.azh.server.datam.page.Page;
import com.azh.server.datam.page.PageNormal;
import com.azh.server.datam.pagecache.PageCache;
import com.azh.server.tx.TxManager;
import com.azh.server.util.ExceptionUtils;
import com.azh.server.util.Parser;
import com.google.common.primitives.Bytes;

/**
 * @author zhaozhenhang <zhaozhenhang@kuaishou.com>
 * Created on 2022-04-25
 * 重做所有崩溃时已完成的commited或者aborted事物
 * 撤销所有崩溃时为完成的active事物
 */
public class Recover {

    private static final byte log_type_insert = 0;

    private static final byte log_type_update = 1;

    // [logType][txId][uid][oldRaw][newRaw]
    private static final int of_type = 0;
    private static final int of_tx_id = of_type + 1;
    private static final int of_update_uid = of_tx_id + 8;
    private static final int of_update_raw = of_update_uid + 8;

    // [logType][txId][pageNumber][offset][raw]
    private static final int of_insert_pageNumber = of_tx_id + 8;
    private static final int of_insert_offset = of_insert_pageNumber + 4;
    private static final int of_insert_raw = of_insert_offset + 2;

    private static final int redo = 0;
    private static final int undo = 1;

    public static byte[] updateLog(long txId, DataItem dataItem) {
        byte[] logType = {log_type_update};
        byte[] txIdRaw = Parser.longToByte(txId);
        byte[] uidRaw = Parser.longToByte(dataItem.getUid());
        byte[] oldRaw = dataItem.getOldRaw();
        SubArray raw = dataItem.getRaw();
        byte[] newRaw = Arrays.copyOfRange(raw.raw, raw.start, raw.end);
        return Bytes.concat(logType, txIdRaw, uidRaw, oldRaw, newRaw);
    }

    /**
     * 恢复数据
     * @param tx
     * @param logger
     * @param pageCache
     */
    public static void recover(TxManager tx, Logger logger, PageCache pageCache) {
        // todo 日志方式得换一下
        System.out.println("Recovering...");
        logger.rewind();
        int maxPageNumber = 0;
        for (; ; ) {
            byte[] log = logger.next();
            if (log == null) {
                break;
            }
            int pageNumber = 0;
            if (isInsertLog(log)) {
                InsertLogInfo info = parserInsertLog(log);
                pageNumber = info.pageNumber;
            }else {
                UpdateLogInfo info = parserUpdateLog(log);
                pageNumber = info.pageNumber;
            }
            if (pageNumber > maxPageNumber) {
                maxPageNumber = pageNumber;
            }
        }
        if (maxPageNumber == 0) {
            maxPageNumber = 1;
        }
        pageCache.truncateBytePageNumber(maxPageNumber);
        System.out.println("truncate to " + maxPageNumber + "pages.");

        redoTx(tx, logger, pageCache);
        System.out.println("redo transactions over");

        undoTx(tx, logger, pageCache);
        System.out.println("undo transactions over");

        System.out.println("recovery over.");
    }

    /**
     * 重做崩溃时已经commit或者aborted的事物
     * @param tx
     * @param logger
     * @param pageCache
     */
    public static void redoTx(TxManager tx, Logger logger, PageCache pageCache) {
        logger.rewind();
        for (; ; ) {
            byte[] log = logger.next();
            if (log == null) {
                break;
            }
            if (isInsertLog(log)) {
                InsertLogInfo insertInfo = parserInsertLog(log);
                long txId = insertInfo.txId;
                if (!tx.isActive(txId)) {
                    // todo 用isActive是否可以换成 isCommitted
                    doInsertLog(pageCache, log, redo);
                }
            }else {
                UpdateLogInfo updateInfo = parserUpdateLog(log);
                long txId = updateInfo.txId;
                if (!tx.isActive(txId)) {
                    doUpdateLog(pageCache, log, redo);
                }
            }
        }
    }

    /**
     * 撤销所有奔溃时为完成的active事物
     * @param tx
     * @param logger
     * @param pageCache
     */
    public static void undoTx(TxManager tx, Logger logger, PageCache pageCache) {
        Map<Long, List<byte[]>> logCache = new HashMap<>();
        logger.rewind();
        for (; ; ) {
            byte[] log = logger.next();
            if (log == null) {
                break;
            }
            if (isInsertLog(log)) {
                InsertLogInfo info = parserInsertLog(log);
                long txId = info.txId;
                if (tx.isActive(txId)) {
                    if (!logCache.containsKey(txId)) {
                        logCache.put(txId, new ArrayList<>());
                    }
                    logCache.get(txId).add(log);
                }
            }else {
                UpdateLogInfo info = parserUpdateLog(log);
                long txId = info.txId;
                if (tx.isActive(txId)) {
                    if (!logCache.containsKey(txId)) {
                        logCache.put(txId, new ArrayList<>());
                    }
                    logCache.get(txId).add(log);
                }
            }
        }

        // 对所有的active log 进行排序（倒序）
        for (Entry<Long, List<byte[]>> entry : logCache.entrySet()) {
            List<byte[]> logs = entry.getValue();
            for (int i = logs.size() - 1; i >= 0; i--) {
                byte[] log = logs.get(i);
                if (isInsertLog(log)) {
                    doInsertLog(pageCache, log, undo);
                }else {
                    doUpdateLog(pageCache, log, undo);
                }
            }
            tx.abort(entry.getKey());
        }
    }

    public static byte[] insertLog(long txId, Page page, byte[] raw) {
        byte[] logTypeRaw = {log_type_insert};
        byte[] txIdRaw = Parser.longToByte(txId);
        byte[] pageNumberRaw = Parser.intToByte(page.getPageNumber());
        byte[] offsetRaw = Parser.shortToByte(PageNormal.getFSO(page));
        return Bytes.concat(logTypeRaw, txIdRaw, pageNumberRaw, offsetRaw, raw);
    }

    private static void doUpdateLog(PageCache pageCache, byte[] log, int status) {
        int pageNumber;
        short offset;
        byte[] raw;
        if (status == redo) {
            UpdateLogInfo updateInfo = parserUpdateLog(log);
            pageNumber = updateInfo.pageNumber;
            offset = updateInfo.offset;
            raw = updateInfo.newRaw;
        }else {
            UpdateLogInfo updateInfo = parserUpdateLog(log);
            pageNumber = updateInfo.pageNumber;
            offset = updateInfo.offset;
            raw = updateInfo.oldRaw;
        }
        Page page = null;
        try {
            page = pageCache.getPage(pageNumber);
        } catch (Exception e) {
            ExceptionUtils.getMessage(e);
        }
        try {
            PageNormal.recoverUpdate(page, raw, offset);
        }finally {
            page.release();
        }
    }

    /**
     * UpdateLogInfo
     * @param log
     * @return
     */
    private static UpdateLogInfo parserUpdateLog(byte[] log) {
        UpdateLogInfo info = new UpdateLogInfo();
        info.txId = Parser.parseLong(Arrays.copyOfRange(log, of_tx_id, of_update_uid));
        long uid = Parser.parseLong(Arrays.copyOfRange(log, of_update_uid, of_update_raw));
        info.offset = (short) (uid & ((1L << 16) - 1));
        uid >>>= 32;
        info.pageNumber = (int) (uid & ((1L << 32) - 1));
        int len = (log.length - of_update_raw) / 2;
        info.oldRaw = Arrays.copyOfRange(log, of_update_raw, of_update_raw + len);
        info.newRaw = Arrays.copyOfRange(log, of_update_raw + len, of_update_raw + len >> 1);
        return info;
    }

    private static void doInsertLog(PageCache pageCache, byte[] log, int status) {
        InsertLogInfo info = parserInsertLog(log);
        Page page = null;
        try {
            page = pageCache.getPage(info.pageNumber);
        } catch (Exception e) {
            ExceptionUtils.getMessage(e);
        }
        try {
            if (status == undo) {
                //将 dataItem 设为无效，逻辑删除
                DataItem.setDataItemRawInvalid(info.raw);
            }
            PageNormal.recoverInsert(page, info.raw, info.offset);
        }finally {
            page.release();
        }
    }

    private static boolean isInsertLog(byte[] log) {
        return log[0] == log_type_insert;
    }

    private static InsertLogInfo parserInsertLog(byte[] log) {
        InsertLogInfo info = new InsertLogInfo();
        info.txId = Parser.parseLong(Arrays.copyOfRange(log, of_tx_id, of_insert_pageNumber));
        info.pageNumber = Parser.parseInt(Arrays.copyOfRange(log, of_insert_pageNumber, of_insert_offset));
        info.offset = Parser.parseShort(Arrays.copyOfRange(log, of_insert_offset, of_insert_raw));
        info.raw = Arrays.copyOfRange(log, of_insert_raw, log.length);
        return info;
    }

    static class InsertLogInfo {
        long txId;
        int pageNumber;
        short offset;
        byte[] raw;
    }

    static class UpdateLogInfo {
        long txId;
        int pageNumber;
        short offset;
        byte[] oldRaw;
        byte[] newRaw;
    }
}
